package com.caius.xiaohashu.comment.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.comment.biz.constant.MQConstants;
import com.caius.xiaohashu.comment.biz.domain.mapper.CommentLikeDOMapper;
import com.caius.xiaohashu.comment.biz.enums.LikeUnLikeCommentTypeEnum;
import com.caius.xiaohashu.comment.biz.model.dto.LikeUnlikeCommentMqDTO;
import com.google.common.collect.Lists;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author Caius
 * @description
 * @since Created in 2025-06-10
 */
@Component
@Slf4j
public class LikeUnlikeComment2DBConsumer {

    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    private DefaultMQPushConsumer consumer;

    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private CommentLikeDOMapper commentLikeDOMapper;

    @Bean(name = "LikeUnlikeComment2DBConsumer")
    public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {
        String group = "xiaohashu_group_" + MQConstants.TOPIC_COMMENT_LIKE_OR_UNLIKE;

        consumer = new DefaultMQPushConsumer();

        consumer.setNamesrvAddr(namesrvAddr);

        consumer.subscribe(MQConstants.TOPIC_COMMENT_LIKE_OR_UNLIKE, "*");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.setMaxReconsumeTimes(3);

        consumer.setConsumeMessageBatchMaxSize(30);

        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            log.info("==> 【评论点赞、取消点赞】本批次消息大小: {}", msgs.size());
            try {
                rateLimiter.acquire();

                List<LikeUnlikeCommentMqDTO> likeUnlikeCommentMqDTOS = Lists.newArrayList();

                msgs.forEach(msg -> {
                    String tag = msg.getTags();
                    String msgJson = new String(msg.getBody());
                    log.info("==> 【评论点赞、取消点赞】Consumer - Tag: {}, Received message: {}", tag, msgJson);

                    likeUnlikeCommentMqDTOS.add(JsonUtils.parseObject(msgJson, LikeUnlikeCommentMqDTO.class));
                });

                Map<Long, List<LikeUnlikeCommentMqDTO>> commentIdAndListMap = likeUnlikeCommentMqDTOS.stream()
                        .collect(Collectors.groupingBy(LikeUnlikeCommentMqDTO::getCommentId));

                List<LikeUnlikeCommentMqDTO> finalLikeUnlikeCommentMqDTOS = Lists.newArrayList();

                commentIdAndListMap.forEach((commentId, ops) -> {
                    // 优化：若某个用户对某评论，多次操作，如点赞 -> 取消点赞 -> 点赞，需进行操作合并，只提取最后一次操作，进一步降低操作数据库的频率
                    Map<Long, LikeUnlikeCommentMqDTO> userLastOp = ops.stream()
                            .collect(Collectors.toMap(
                                    LikeUnlikeCommentMqDTO::getUserId,
                                    Function.identity(),
                                    // 合并策略：当出现重复键（同一用户多次操作）时，保留时间更晚的记录
                                    (oleValue, newValue) ->
                                            oleValue.getCreateTime().isAfter(newValue.getCreateTime()) ? oleValue : newValue

                            ));
                    finalLikeUnlikeCommentMqDTOS.addAll(userLastOp.values());
                });

                // 批量操作数据库
                executeBatchSQL(finalLikeUnlikeCommentMqDTOS);

                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                log.error("", e);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        });

        consumer.start();
        return consumer;
    }

    /**
     * 批量操作数据库
     * @param values
     */
    private void executeBatchSQL(List<LikeUnlikeCommentMqDTO> values) {
        List<LikeUnlikeCommentMqDTO> likes = values.stream()
                .filter(op -> Objects.equals(op.getType(), LikeUnLikeCommentTypeEnum.LIKE.getCode()))
                .toList();

        List<LikeUnlikeCommentMqDTO> unlikes = values.stream()
                .filter(op -> Objects.equals(op.getType(), LikeUnLikeCommentTypeEnum.UNLIKE.getCode()))
                .toList();

        if (CollUtil.isNotEmpty(likes)) {
            commentLikeDOMapper.batchInsert(likes);
        }

        if (CollUtil.isNotEmpty(unlikes)) {
            commentLikeDOMapper.batchDelete(unlikes);
        }
    }

    @PreDestroy
    public void destory() {
        if (Objects.nonNull(consumer)) {
            try {
                consumer.shutdown();
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }
}
